package net.sf.calibur.transport.mina;

import java.io.InputStream;

import net.sf.calibur.transport.CipherFactory;
import net.sf.calibur.transport.MessageReceiver;
import net.sf.calibur.transport.Session;
import net.sf.calibur.transport.TransportException;
import net.sf.calibur.transport.message.Message;
import net.sf.calibur.transport.message.MessageHead;

import org.apache.mina.core.buffer.IoBuffer;

public class IoBufferMessageReceiver implements MessageReceiver<IoBuffer> {
	enum Phase {
		recv_message_size, recv_message_body
	};

	// 最大消息长度，用于验证收到的消息长度是否合法
	public final static int MAX_MESSAGE_SIZE = 4096;

	// 当前消息解析状态
	private Phase currentPhase;
	// 当前消息长度
	private int messageSize;
	// 当前绑定的session
	private Session session;

	public IoBufferMessageReceiver(Session session) {
		this.session = session;
		currentPhase = Phase.recv_message_size;
		messageSize = 0;
	}

	public void clear() {
		currentPhase = Phase.recv_message_size;
		messageSize = 0;
		session = null;
	}

	private Message decodeMessage(IoBuffer buffer) throws Exception {
		int sequ = -1;
		
		/*
		 * MessageType 1字节
		 * 最高位（第7位），表示应用数据是否加密。
		 * 第4-6位，报文头类型，000无报文头，001有一个long型的clientTimestamp
		 * 第0-3位，报文类型，参考Message里的定义
		 */
		byte messageType = buffer.get();
		// boolean crypt = session.isCrypto() || ((messageType & 0x80) != 0);
		boolean crypt = (messageType & 0x80) != 0;
		int headType = (messageType & 0x70) >> 4;
		messageType &= 0x0f;

		InputStream stream = buffer.asInputStream();
		MessageHead head = MessageHead.readMessageHead(headType, stream);
		Message message = Message.createMessage(messageType, head, 
				session.getSessionManager().getCommandFactory());

		if (messageType == Message.MT_APPLICATION) {
			sequ = session.getReceiveSequence();
			if (crypt) {
				stream = CipherFactory.getInstance().getCipherInputStream(
						session, buffer.asInputStream());
			}

		}
		message.deserialize(stream);
		message.setSequence(sequ);

		if (messageType == Message.MT_APPLICATION) {
			session.incReceiveSequence();
		}

		return message;
	}

	@Override
	public Message onDataArrived(IoBuffer buffer) throws Exception {
		if (currentPhase == Phase.recv_message_body) {
			if (buffer.remaining() < messageSize) {
				return null;
			} else {
				try {
					return decodeMessage(buffer);
				} finally {
					currentPhase = Phase.recv_message_size;
				}
			}
		} else if (currentPhase == Phase.recv_message_size) {
			if (buffer.remaining() < 4) {
				return null;
			} else {
				messageSize = buffer.getInt();
				if (messageSize <= 0) {
					throw new TransportException(String.format(
							"Suspicious message size %d came from %s.",
							messageSize, session.getRemoteAddress().toString()));
				} else if (messageSize > MAX_MESSAGE_SIZE) {
					throw new TransportException(String.format(
							"Message size %d exceeds limit %d came from %s.",
							messageSize, MAX_MESSAGE_SIZE, session
									.getRemoteAddress().toString()));
				} else {
					try {
						currentPhase = Phase.recv_message_body;
						return onDataArrived(buffer);
					} catch (TransportException te) {
						throw te;
					} catch (Throwable t) {
						throw new TransportException(String.format(
								"deserialize message came from %s failed.",
								session.getRemoteAddress().toString()), t);
					}
				}
			}
		} else {
			throw new IllegalStateException("Error message receive phase "
					+ currentPhase);
		}
	}
}
